package com.cictec.middleware.elasticsearch.camel.processor;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.cictec.middleware.commons.utils.DateUtils;
import com.cictec.middleware.elasticsearch.service.ElasticsearchPositionService;

/**
 * 中间件实时计算 进出站信息保存
 * 
 * @author ql
 */
@SuppressWarnings("rawtypes")
@Component
public class MiddlewareTimingPointTMESProcessor implements Processor {

	private static Logger logger = LoggerFactory.getLogger(MiddlewareTimingPointTMESProcessor.class);

	@Autowired
	private ElasticsearchPositionService elasticsearchPositionService;
	
	@Override
	public void process(Exchange exchange) throws Exception {
		Message message = exchange.getIn();
		Map data = (Map) message.getBody();
		// 实时数据
		realCaculateBatchHandler(data);
	}

	/*
	 * ******* realCaculate start ****************/
	private List<Map> realCaculateBatchList = new ArrayList<Map>();
	/**
	 * 当前实时数据刷新时刻
	 */
	private Date realTimer = null;
	private boolean isSaveMonitorStart = false;

	/**
	 * 批量保存实时数据
	 * 
	 * @param data
	 */
	private void realCaculateBatchHandler(Map data) {
		realTimer = new Date();
		if (!isSaveMonitorStart) {
			// 开启实时数据批量入库监控
			isSaveMonitorStart = true;
			startSaveMonitor();
		}

		realCaculateBatchList.add(data);

//		if (realCaculateBatchList.size() > 300) {
		if (realCaculateBatchList.size() > 1) {
			// 入库一次
			realCaculateBatchHandler();
		}
		System.out.println("\n进出站："+JSON.toJSONString(data));
	}

	private void realCaculateBatchHandler(){
		logger.debug("保存ES 批量实时进出站信息");

		elasticsearchPositionService.saveEnterOutStations(realCaculateBatchList, null);
		realCaculateBatchList = new ArrayList<Map>();// 重置缓存

	}
	
	/**
	 * 监控批量写入ES任务
	 */
	private void startSaveMonitor() {
		new Thread() {
			public void run() {
				while (true) {

					Date nowTime = new Date();
					if (DateUtils.secondInterval(nowTime, realTimer) > 10 * 60) {
						if (realCaculateBatchList.size() > 0) {
							realCaculateBatchHandler();
						}
					}

					try {
						int seconds = 10 * 60; // 十分钟内没有数据，更新数据
						Thread.sleep(seconds * 1000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		}.start();
	}

}
